package cn.com.scooper.impl.netty;

import cn.com.scooper.common.constant.CommonConstant;
import cn.com.scooper.dao.xf.MineIotDeviceDataDao;
import cn.com.scooper.impl.ks.KSReportDataHandler;
import cn.com.scooper.impl.ks.PropertyReader;
import cn.com.scooper.pojo.dto.ks.ReportRequestPkgDto;
import cn.com.scooper.pojo.po.MineIotDeviceDataPo;
import com.alibaba.fastjson.JSONObject;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PreDestroy;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
@Slf4j
public class ReportNettyClient {

    // netty 工作线程池
    private EventLoopGroup group;

    // 通道
    private Channel channel;

    private KSReportDataHandler ksReportDataHandler;

    private ReportNettyClient reportNettyClient;

    @Autowired
    private PropertyReader propertyReader;

    @Autowired
    private MineIotDeviceDataDao mineIotDeviceDataDao;

    private void initClient() {
        group = new NioEventLoopGroup();
        reportNettyClient = this;
        doConnect();
    }

    public void doConnect(){

        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>(){
                    @Override
                    protected void initChannel(SocketChannel socketChannel) {
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        //拆包 间隔符
                        ByteBuf delimiter = Unpooled.copiedBuffer(CommonConstant.REPORT_DELIMITER.getBytes());
                        //2048000 和服务端保持一次，不要修改
                        pipeline.addLast(new DelimiterBasedFrameDecoder(2048000,delimiter));
                        // 客户端的逻辑
                        pipeline.addLast(new IdleStateHandler(120, 120, 0, TimeUnit.MINUTES));
                        pipeline.addLast(new ClientHandler(ksReportDataHandler,reportNettyClient));
                    }
                });
        try {
            ChannelFuture future = bootstrap.connect(propertyReader.getNettyIp(),propertyReader.getNettyPort());
            future.addListener((ChannelFuture futureListener) -> {
                if (!futureListener.isSuccess()) {
                    //连接tcp服务器不成功 30后重连
                    log.info("【向市局上报数据->建立netty连接失败】在{}秒之后尝试重连{}!", 30, propertyReader.getNettyIp()+":"+propertyReader.getNettyPort());
                    futureListener.channel().eventLoop().schedule(() -> doConnect(), 30, TimeUnit.SECONDS);
                    // 连接失败，检查队列中是否有暂存未上报的数据
                    log.info("【向市局上报数据->建立netty连接失败】若队列中存在未上报的数据则暂存数据库");
                    saveDb();
                }else{
                    channel = futureListener.channel();
                    log.info("【向市局上报数据->建立netty连接成功】连接：{}", propertyReader.getNettyIp()+":"+propertyReader.getNettyPort());
                    // 连接成功，检查数据库中是否有暂存未上报的数据
                    log.info("【向市局上报数据->建立netty连接成功】若数据库中存在未上报的数据则取出存入队列");
                    MineIotDeviceDataPo mineIotDeviceDataQo = new MineIotDeviceDataPo();
                    mineIotDeviceDataQo.setIsReported((byte)0);
                    List<MineIotDeviceDataPo> mineIotDeviceDataList = mineIotDeviceDataDao.selectAll(mineIotDeviceDataQo);
                    if( !mineIotDeviceDataList.isEmpty() ){
                        for (MineIotDeviceDataPo mineIotDeviceDataPo : mineIotDeviceDataList) {
                            ksReportDataHandler.offer(JSONObject.parseObject(mineIotDeviceDataPo.getReportData(), ReportRequestPkgDto.class),mineIotDeviceDataPo.getReportEnterprise());
                        }
                        List<Integer> reportedIds = mineIotDeviceDataList.stream().map(s->s.getId()).collect(Collectors.toList());
                        mineIotDeviceDataDao.updateReported(reportedIds);
                    }
                }
            });
        } catch (Exception e) {
            log.error("【向市局上报数据->建立netty连接失败】{}", e.getMessage());
        }
    }

    /**
     * 将队列中暂存未上报的数据存入数据库
     */
    private void saveDb() {
        ReportRequestPkgDto reportRequestPkgDto;
        MineIotDeviceDataPo mineIotDeviceDataPo = new MineIotDeviceDataPo();
        mineIotDeviceDataPo.setIsReported((byte)0);
        mineIotDeviceDataPo.setGmtCreate(new Date());
        mineIotDeviceDataPo.setGmtUpdate(new Date());
        while ((reportRequestPkgDto=ksReportDataHandler.poll(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_HSSN))!=null){
            mineIotDeviceDataPo.setReportData(JSONObject.toJSONString(reportRequestPkgDto));
            mineIotDeviceDataPo.setReportEnterprise(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_HSSN);
            mineIotDeviceDataDao.insert(mineIotDeviceDataPo);
        }
        while ((reportRequestPkgDto=ksReportDataHandler.poll(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_NFSN))!=null){
            mineIotDeviceDataPo.setReportData(JSONObject.toJSONString(reportRequestPkgDto));
            mineIotDeviceDataPo.setReportEnterprise(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_NFSN);
            mineIotDeviceDataDao.insert(mineIotDeviceDataPo);
        }
        while ((reportRequestPkgDto=ksReportDataHandler.poll(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_CYKY))!=null){
            mineIotDeviceDataPo.setReportData(JSONObject.toJSONString(reportRequestPkgDto));
            mineIotDeviceDataPo.setReportEnterprise(CommonConstant.ReportEnterpriseName.KS_ENTERPRISE_CYKY);
            mineIotDeviceDataDao.insert(mineIotDeviceDataPo);
        }
    }

    /**
     * @param req
     *
     * 示例：
     * {
     * 	"appId": "3c9a6868a6d74e348708ad3f0c15c25b",
     * 	"serviceId": "KS_REPORT_ACTUAL_DATA",
     * 	"dataId": "1563259577529",
     * 	"data": "BLnaeCX2aOtX79+6c8z X6zycGksa5YxUQyA2SUzyhCkRZoc5vUjvGmBuOEeWL/NrO2bJUHfAqFFO6Do GzPs5eQ==",
     * 	"dataType": "01"
     * }@@
     *
     */
    public void writeMsg(byte[] req){
        log.info("【向市局上报数据->客户端写入数据】");
        ByteBuf message = Unpooled.buffer(req.length);
        message.writeBytes(req);
        channel.writeAndFlush(message);
    }

    @PreDestroy
    public void closeConnect() {
        log.info("【向市局上报数据->客户端netty连接关闭】项目已停止运行");
        saveDb();
        channel.close();
        group.shutdownGracefully();
    }

    public void setKsReportDataHandler(KSReportDataHandler ksReportDataHandler) {
        this.ksReportDataHandler = ksReportDataHandler;
        initClient();
    }

    public boolean isConnected() {
        return channel != null && channel.isActive();
    }
}
